#include "./stdafx.h" 
#include "./my_mqtt.h"


 
using namespace std;
 

#include "./rapidjson/rapidjson.h" 
#include "./rapidjson/document.h" 
#include "./rapidjson/writer.h" 
#include "./rapidjson/stringbuffer.h" 
#include "./rapidjson/prettywriter.h" 
#include "./rapidjson/filereadstream.h"
#include "./rapidjson/filewritestream.h"

#include "./mqtt_msg.h"


int    my_mqtt_on_connect_ok = 0; 
int 	my_onsend_finished = 0;
int 	my_messageArrivedcount = 0;
int 	my_deliveryCompleted_count = 0;

 
 
//int myMsgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message);

CMyMQTTAsync::CMyMQTTAsync()
{
	my_mqtt_on_connect_ok = 0;

	m_my_result_val = 0;
}


CMyMQTTAsync::~CMyMQTTAsync()
{
	disconnect();
}


void myOnConnect(void* context, MQTTAsync_successData* response)
{  
	my_mqtt_on_connect_ok = 1; 
	 
}
 
void myOnDisconnect(void* context, MQTTAsync_successData* response)
{
	printf("Successful disconnection\n");
	  
}


void myOnConnectFailure(void* context, MQTTAsync_failureData* response)
{
	printf("Connect failed, rc %d\n", response ? response->code : 0); 
}


void myDeliveryComplete(void* context, MQTTAsync_token dt)
{  
	//printf("Delivery complete for token %d \n", dt);
	++my_deliveryCompleted_count; 

	//  MQTTAsync_setCallbacks�������ص����� �� context ����setCallbacks����ģ� ����û�а�DeliveryComplete��connlostҲ�ŵ����棨�򵥴�����
} 


void myconnlost(void *context, char *cause)
{ 
	//  MQTTAsync_setCallbacks�������ص����� �� context ����setCallbacks����ģ� ����û�а�DeliveryComplete��connlostҲ�ŵ����棨�򵥴�����
	printf("\n myMqtt Connection lost  cause: %s\n",  cause); 
 
	printf("nmyMqtt Reconnecting\n");
	 
	 
}

void _disconnect(void* context)
{ 
	MQTTAsync client = (MQTTAsync)context;

	MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
	int rc;  
	 
	opts.onSuccess = myOnDisconnect;
	opts.context = client; 

	//printf(" disconnect 11111\n" );  
	if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start sendMessage, return code %d\n", rc);
		exit(EXIT_FAILURE);
	}  
	//printf(" disconnect 22222\n" );  
	MQTTAsync_destroy(&client); 
	//printf(" disconnect 3333\n" );  
}


void CMyMQTTAsync::disconnect()
{ 
	_disconnect(m_client); 
}	 

bool  CMyMQTTAsync::connect_mqtt(char* pub_topic_clientID, char * mqtt_svr_addr,  MQTTAsync_messageArrived *  my_fun, void* context)
{
	if( strlen( mqtt_svr_addr ) == 0 ) return false; 
 
	my_mqtt_on_connect_ok = 0; 
  	my_onsend_finished = 0;
  	my_messageArrivedcount = 0;
  	my_deliveryCompleted_count = 0;
    
	SetResultVal(false);

	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	  
	MQTTAsync_create(&m_client, mqtt_svr_addr, pub_topic_clientID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
	
	if( context == NULL ) 
		MQTTAsync_setCallbacks(m_client, this, myconnlost,  my_fun, myDeliveryComplete);
	else
		MQTTAsync_setCallbacks(m_client, context, myconnlost,  my_fun, myDeliveryComplete);
	
	int rc;

	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = myOnConnect;
	conn_opts.onFailure = myOnConnectFailure;
	conn_opts.context = m_client;
	if ((rc = MQTTAsync_connect(m_client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed1 to start connect, return code %d\n", rc);
		return false;
	} 
	
	int   wait_seconds = 5; 
	while ( !my_mqtt_on_connect_ok )
	{ 
		if(wait_seconds-- <= 0) break;
		
		// base::MySleep(1000L); 
		sleep(100);
	}
	
	if( !my_mqtt_on_connect_ok ) {
		printf("MQTTAsync_connect Failed \n");
		return false;
	}

	printf("pub_topic: connect success \n" );  
	
	return true;	 
}


bool CMyMQTTAsync::analyse_confirm( const char * json )
{ 
    	printf("-----��ȡ  Json����:\n"); 
    //���ļ��ж�ȡ��ע�������ֿ�������ȷ���ļ��Ƿ�������ϣ������ȡ���������⣩ 
	char   chrStatusDesc[32] = {0};
        int    statusCode = -1;
	string strToken;

	try
	{
		rapidjson::Document newDoc; 
		newDoc.Parse(json);

		//�жϽ��������ж�ȡ���ַ����Ƿ��д��� 
		if (newDoc.HasParseError()) {
			printf("Json Parse error:%d \n", newDoc.GetParseError()); 
			return false;
		}
		else {  
			if (newDoc.HasMember("token")) { 
				strToken = newDoc["token"].GetString() ;
				printf("token:%s \n", strToken.c_str());  
			}
			else {
				printf(" [token] is not exist \n" );  
				return false;
			}  

			if (newDoc.HasMember("statusCode")) {
				statusCode = newDoc["statusCode"].GetInt();
				printf("statusCode:%d \n", statusCode);  
			}
			else {
				printf(" [statusCode] is not exist \n" );  
				return false;
			}  

			if (newDoc.HasMember("statusDesc")) {
				strcpy(chrStatusDesc, newDoc["statusDesc"].GetString()); 
				printf("statusDesc:%s \n", chrStatusDesc);  
			}
			else {
				printf(" [statusDesc] is not exist \n" );  
				return false;
			} 

			if(   statusCode== 0 && strcmp(chrStatusDesc, "OK")==0 ) 
				return true;
			else
				return false;
		}
      }
    catch(...) {
		printf("unknown error: try catch"); 
    } 

   return false;
}
 

void CMyMQTTAsync::SetResultVal(bool b)
{
	//m_subsrcibe_threadLock.Lock();
	m_my_result_val = b; 
	//m_subsrcibe_threadLock.UnLock();
}
 
void CMyMQTTAsync::send_string( int qos, char* pTopic,  char * msg_data )
{
	if( msg_data == NULL ) return;
	if( strlen(msg_data) <= 0 ) return;
	 
	printf("send msg: pTopic = %s, len=%d \n", pTopic, strlen(msg_data));
	
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	      
		 
	pubmsg.payload = (void*)msg_data; 
	pubmsg.payloadlen = strlen( msg_data );  
	pubmsg.retained = 0;
	pubmsg.qos = qos; 	
	//opts.onSuccess =  onSend;
	opts.context = m_client;  

	int rc = MQTTAsync_send(m_client, pTopic, pubmsg.payloadlen, pubmsg.payload,pubmsg.qos, pubmsg.retained, &opts);
	if ( rc  != MQTTASYNC_SUCCESS )
	{
		printf("Failed to sendMessage, return code %d\n", rc);
		return;
	} 
  
}
 


void CMyMQTTAsync::send_msg( int qos, char* pTopic, char * msg_data, unsigned short msg_len )
{
	if( msg_data == NULL ) return;
	if( strlen(msg_data) <= 0 ) return;
	 
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	     
	int last_idx =  strlen(msg_data); 
		 
	pubmsg.payload = (void*)msg_data;
	pubmsg.payloadlen = msg_len;  
	pubmsg.retained = 0;
	pubmsg.qos = qos;
		
	//opts.onSuccess =  onSend;
	opts.context = m_client; 

	int rc = MQTTAsync_send(m_client, pTopic, pubmsg.payloadlen, pubmsg.payload,pubmsg.qos, pubmsg.retained, &opts);
	if ( rc  != MQTTASYNC_SUCCESS )
	{
		printf("Failed to sendMessage, return code %d\n", rc);
		//return;
	} 
	 
}
 

 

bool  CMyMQTTAsync::send_msg(char* topic, char* msg, unsigned short msg_len, unsigned char msg_format )
{
	if( msg == NULL ) return false; 

	if( msg_format == TOPIC_PUB_DFORMAT_TYPE_JSON )
		send_string(QOS, topic, msg);
	else
		send_msg(QOS, topic, msg, msg_len);

	return true;
}


void  CMyMQTTAsync::subscribe_topic(char* topic)
{
	if( topic == NULL ) return; 

	m_arySubTopic.clear();  
	m_arySubTopic.push_back( topic ); 
	int rc;

	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer; 
	for (int i=0; i<m_arySubTopic.size(); i++ ) {  
        
		string  topicName = m_arySubTopic[i];
		
 
		if( topicName.length() <= 0 ) continue; 
   
		if ((rc = MQTTAsync_subscribe(m_client, topicName.c_str(), QOS, &opts)) != MQTTASYNC_SUCCESS)
		{
			printf("subscribe Topic=%s: Failed to start subscribe, return code %d\n", topicName.c_str(), rc);
			//exit(EXIT_FAILURE);
			continue;
		} 

		// base::MySleep(50);
		sleep(50);
    }   
}

bool  CMyMQTTAsync::IsCurrentTopic(char* topic )
{
	for (int i=0; i<m_arySubTopic.size(); i++ ) {  
        
		string  topicName = m_arySubTopic[i];
		
		if( topicName.length() <= 0 ) continue; 
	  
		 
		if( strcmp(topic, topicName.c_str()) == 0 ) return true;
    }  

	return false;
}
 
 /*
string gettimestring()
{ 
	time_t tt = time(NULL);   
	struct tm   * p = localtime(&tt); 
	
	unsigned short minsec = 0;

#ifdef WIN32
	SYSTEMTIME				stime;  
	GetLocalTime(&stime);

	minsec = (unsigned short)(stime.wMilliseconds / 1000);
#else
	struct timeval  tv;
	gettimeofday(&tv, NULL);

	minsec = (unsigned short)(tv.tv_usec / 1000);
#endif

	char buf[32] = {0}; 

	sprintf(buf, "%d-%02d-%02dT%02d:%02d:%02d.%03Z", p->tm_year, p->tm_mon+1, p->tm_mday, p->tm_hour, p->tm_min, p->tm_sec, minsec); 
 
	return buf;
}*/
 


bool CMyMQTTAsync::Send_Reply_pack(char * topip, char * chrToken, int statusCode, char* statusDesc,  char * errorDesc, char* errorTag)
{ 

	char   pResultData[1024];
	memset(pResultData, 0, 1024);
	sprintf(pResultData,  "{ \"token\":\"%s\" , \"timestamp\":  \"%d\", \"statusCode\": %d, \"statusDesc\":\"%s\"", chrToken, time(NULL), statusCode, statusDesc);
 

	char chrError[512] = {0};
	if( strncpy(statusDesc, "OK", 0) != 0 )
	{
		if( errorDesc != NULL && errorTag != NULL ) {
			sprintf(chrError,  "{ \"body\" : { \"errorDesc\":\"%s\", \"errorTag\":\"%s\" }  }", errorDesc, errorTag);
		}
		else {
			sprintf(chrError,  "{ \"body\" : { \"errorDesc\":\"\", \"errorTag\":\"\" }  }", errorDesc, errorTag);
		}
		strcat(pResultData, ",");
		strcat(pResultData, chrError);
	}

	strcat(pResultData, "  }"); 

	send_string(QOS, topip, pResultData);

	return true;
}
